package com.wejoy.video.middle.station.data.event.wjstream;

import com.wejoy.video.middle.station.data.exception.BusinessException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.dao.DataAccessResourceFailureException;
import org.springframework.data.redis.connection.stream.Consumer;
import org.springframework.data.redis.connection.stream.ReadOffset;
import org.springframework.data.redis.connection.stream.Record;
import org.springframework.data.redis.connection.stream.StreamOffset;
import org.springframework.data.redis.stream.StreamListener;
import org.springframework.data.redis.stream.StreamMessageListenerContainer.ConsumerStreamReadRequest;
import org.springframework.data.redis.stream.StreamMessageListenerContainer.StreamReadRequest;
import org.springframework.data.redis.stream.Task;
import org.springframework.util.ErrorHandler;

import java.time.Duration;
import java.util.List;
import java.util.Optional;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.function.BiFunction;
import java.util.function.Predicate;

/**
 * (本文件源码仅改动了doLoop方法中的代码)
 * @param <K>
 * @param <V>
 */
public class WjStreamPollTask<K, V extends Record<K, ?>> implements Task {
    private final static Logger logger = LoggerFactory.getLogger(WjStreamPollTask.class);
    private final StreamReadRequest<K> request;
    private final StreamListener<K, V> listener;
    private final ErrorHandler errorHandler;
    private final Predicate<Throwable> cancelSubscriptionOnError;
    private final BiFunction<K, ReadOffset, List<V>> readFunction;

    private final WjStreamPollTask.PollState pollState;
    private volatile boolean isInEventLoop = false;

    WjStreamPollTask(StreamReadRequest<K> streamRequest, StreamListener<K, V> listener, ErrorHandler errorHandler,
                   BiFunction<K, ReadOffset, List<V>> readFunction) {

        this.request = streamRequest;
        this.listener = listener;
        this.errorHandler = Optional.ofNullable(streamRequest.getErrorHandler()).orElse(errorHandler);
        this.cancelSubscriptionOnError = streamRequest.getCancelSubscriptionOnError();
        this.readFunction = readFunction;
        this.pollState = createPollState(streamRequest);
    }

    private static WjStreamPollTask.PollState createPollState(StreamReadRequest<?> streamRequest) {

        StreamOffset<?> streamOffset = streamRequest.getStreamOffset();

        if (streamRequest instanceof ConsumerStreamReadRequest) {
            return WjStreamPollTask.PollState.consumer(((ConsumerStreamReadRequest<?>) streamRequest).getConsumer(), streamOffset.getOffset());
        }

        return WjStreamPollTask.PollState.standalone(streamOffset.getOffset());
    }

    /*
     * (non-Javadoc)
     * @see org.springframework.data.redis.stream.Cancelable#cancel()
     */
    @Override
    public void cancel() throws DataAccessResourceFailureException {
        this.pollState.cancel();
    }

    /*
     * (non-Javadoc)
     * @see org.springframework.data.redis.stream.Task#getState()
     */
    @Override
    public State getState() {
        return pollState.getState();
    }

    /*
     * (non-Javadoc)
     * @see org.springframework.data.redis.stream.Task#awaitStart(java.time.Duration)
     */
    @Override
    public boolean awaitStart(Duration timeout) throws InterruptedException {
        return pollState.awaitStart(timeout.toNanos(), TimeUnit.NANOSECONDS);
    }

    /*
     * (non-Javadoc)
     * @see org.springframework.scheduling.SchedulingAwareRunnable#isLongLived()
     */
    @Override
    public boolean isLongLived() {
        return true;
    }

    /*
     * (non-Javadoc)
     * @see java.lang.Runnable#run()
     */
    @Override
    public void run() {

        pollState.starting();

        try {

            isInEventLoop = true;
            pollState.running();
            doLoop(request.getStreamOffset().getKey());
        } finally {
            isInEventLoop = false;
        }
    }

    private void doLoop(K key) {
        String listenerName = listener.toString();
        do {
            try {
                // allow interruption
                // logger.info("{} 准备sleep", listenerName);
                Thread.sleep(0);
                // logger.info("{} 获取到线程", listenerName);
                List<V> read = readFunction.apply(key, pollState.getCurrentReadOffset());
                for (V message : read) {
                    consumeMessage(message, 1);
                    pollState.updateReadOffset(message.getId().getValue());
                }
            } catch (InterruptedException e) {
                 // WJ注释了代码
//                cancel();
                logger.error("{} 发生了InterruptedException异常", listenerName);
                logger.error("发生了InterruptedException异常", e);
                Thread.currentThread().interrupt();
            } catch (RuntimeException e) {
                logger.error("{} 发生了RuntimeException异常", listenerName);
                logger.error("发生了RuntimeException异常", e);
                // WJ注释了代码
//                if (cancelSubscriptionOnError.test(e)) {
//                    cancel();
//                }
                errorHandler.handleError(e);
            }
        } while (pollState.isSubscriptionActive());
    }

    /**
     * 消费消息
     * 1.正常进行消费
     * 2.如果消费者发生异常, 则重试, 重试最多3次
     * 3.每次重试延迟200毫秒
     * @param message
     * @param times
     */
    private void consumeMessage(V message, int times) {
        // 最大执行次数
        if (times > 3) {
            throw new BusinessException("重试了" + times + "次, 丢弃消息");
        }
        String listenerName = listener.toString();
        try {
            listener.onMessage(message);
        } catch (RuntimeException e) {
            logger.error("消费者 {}发生了异常 {}", listenerName, times);
            logger.error("消费者发生了异常", e);
            throw new BusinessException("消费异常");
//            try {
//                Thread.sleep(200);
//            } catch (InterruptedException interruptedException) {
//                interruptedException.printStackTrace();
//            }
            // consumeMessage(message, times + 1);
        }
    }

    @Override
    public boolean isActive() {
        return State.RUNNING.equals(getState()) || isInEventLoop;
    }

    /**
     * Object representing the current polling state for a particular stream subscription.
     */
    static class PollState {

        private final ReadOffsetStrategy readOffsetStrategy;
        private final Optional<Consumer> consumer;
        private volatile ReadOffset currentOffset;
        private volatile State state = State.CREATED;
        private volatile CountDownLatch awaitStart = new CountDownLatch(1);

        private PollState(Optional<Consumer> consumer, ReadOffsetStrategy readOffsetStrategy, ReadOffset currentOffset) {

            this.readOffsetStrategy = readOffsetStrategy;
            this.currentOffset = currentOffset;
            this.consumer = consumer;
        }

        /**
         * Create a new state object for standalone-read.
         *
         * @param offset the {@link ReadOffset} to use.
         * @return new instance of {@link WjStreamPollTask.PollState}.
         */
        static WjStreamPollTask.PollState standalone(ReadOffset offset) {

            ReadOffsetStrategy strategy = ReadOffsetStrategy.getStrategy(offset);
            return new WjStreamPollTask.PollState(Optional.empty(), strategy, strategy.getFirst(offset, Optional.empty()));
        }

        /**
         * Create a new state object for consumergroup-read.
         *
         * @param consumer the {@link Consumer} to use.
         * @param offset the {@link ReadOffset} to apply.
         * @return new instance of {@link WjStreamPollTask.PollState}.
         */
        static WjStreamPollTask.PollState consumer(Consumer consumer, ReadOffset offset) {

            ReadOffsetStrategy strategy = ReadOffsetStrategy.getStrategy(offset);
            Optional<Consumer> optionalConsumer = Optional.of(consumer);
            return new WjStreamPollTask.PollState(optionalConsumer, strategy, strategy.getFirst(offset, optionalConsumer));
        }

        boolean awaitStart(long timeout, TimeUnit unit) throws InterruptedException {
            return awaitStart.await(timeout, unit);
        }

        public State getState() {
            return state;
        }

        /**
         * @return {@literal true} if the subscription is active.
         */
        boolean isSubscriptionActive() {
            return state == State.STARTING || state == State.RUNNING;
        }

        /**
         * Set the state to {@link org.springframework.data.redis.stream.Task.State#STARTING}.
         */
        void starting() {
            state = State.STARTING;
        }

        /**
         * Switch the state to {@link org.springframework.data.redis.stream.Task.State#RUNNING}.
         */
        void running() {

            state = State.RUNNING;

            CountDownLatch awaitStart = this.awaitStart;

            if (awaitStart.getCount() == 1) {
                awaitStart.countDown();
            }
        }

        /**
         * Set the state to {@link org.springframework.data.redis.stream.Task.State#CANCELLED} and re-arm the
         * {@link #awaitStart(long, TimeUnit) await synchronizer}.
         */
        void cancel() {

            awaitStart = new CountDownLatch(1);
            state = State.CANCELLED;
        }

        /**
         * Advance the {@link ReadOffset}.
         */
        void updateReadOffset(String messageId) {
            currentOffset = readOffsetStrategy.getNext(getCurrentReadOffset(), consumer, messageId);
        }

        ReadOffset getCurrentReadOffset() {
            return currentOffset;
        }
    }
}
